package com.stoneframe.kafkaserver;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class KafkaConsumer extends Thread {
	
	private final ConsumerConnector consumer;
	private final String topic;
	public KafkaConsumer(String topic){
		consumer=Consumer.createJavaConsumerConnector(
				createConsumerConfig());
		this.topic=topic;
	}
	
	private ConsumerConfig createConsumerConfig(){
		Properties props=new Properties();
		props.put("zookeeper.connect", KafkaProperties.zkConnect);
		props.put("group.id", KafkaProperties.groupId);
		props.put("zookeeper.session.timeout.ms", "40000");
		props.put("zookeeper.sync.time.ms", "200");
		props.put("auto.commit.interval.ms", "1000");
		return new ConsumerConfig(props);
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		Map<String, Integer> topicCountMap=new HashMap();
		topicCountMap.put(topic, new Integer(1));
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap=
				consumer.createMessageStreams(topicCountMap);
		KafkaStream<byte[], byte[]> stream=consumerMap.get(topic).get(0);
		ConsumerIterator<byte[], byte[]> iterator=stream.iterator();
		while(iterator.hasNext()){
			System.out.println("receive：" + new String(iterator.next().message()));
			try{
				sleep(3000);
			}catch(InterruptedException e){
				e.printStackTrace();
			}
		}
	}
	
	

}
